package al.xc.redis;

import al.xc.common.task.TaskThread;
import al.xc.common.util.SingleUtils;
import lombok.extern.slf4j.Slf4j;
import redis.clients.jedis.JedisPubSub;

import java.util.concurrent.atomic.AtomicInteger;

/**
 * 发布订阅事件监听
 */
@Slf4j
public class RedisPubSubImpl {

    /**
     * 处理线程
     */
    private TaskThread thread_;

    /**
     * 监听中 0 未监听 1 频道 2 模式
     */
    private final AtomicInteger listen_ = new AtomicInteger(0);

    /**
     * 事件监听器
     */
    private IRedisPubSubHandler handler_;

    /**
     * jedis订阅发布
     */
    private JedisPubSub pubSub_;

    public RedisPubSubImpl(IRedisPubSubHandler handler) {
        handler_ = handler;
        thread_ = new TaskThread();
        thread_.initialize(1, this.getClass().getName());

        pubSub_ = new JedisPubSub() {
            @Override
            public void onMessage(String channel, String message) {
                if (null != handler_) {
                    handler_.onMessage(channel, message);
                }
            }

            @Override
            public void onPMessage(String pattern, String channel, String message) {
                if (null != handler_) {
                    handler_.onPMessage(pattern, channel, message);
                }
            }

            @Override
            public void onSubscribe(String channel, int subscribedChannels) {
                if (null != handler_) {
                    handler_.onSubscribe(channel, subscribedChannels);
                }
            }

            @Override
            public void onUnsubscribe(String channel, int subscribedChannels) {
                if (null != handler_) {
                    handler_.onUnsubscribe(channel, subscribedChannels);
                }
            }

            @Override
            public void onPUnsubscribe(String pattern, int subscribedChannels) {
                if (null != handler_) {
                    handler_.onPUnsubscribe(pattern, subscribedChannels);
                }
            }

            @Override
            public void onPSubscribe(String pattern, int subscribedChannels) {
                if (null != handler_) {
                    handler_.onPSubscribe(pattern, subscribedChannels);
                }
            }
        };
    }

    protected void finalize() {
        unsubscribe();
        punsubscribe();
        thread_.unInitialize();
    }

    /**
     * 订阅频道
     * @param channel 频道名称数组
     */
    public boolean subscribe(String... channel) {
        synchronized (listen_) {
            if (!listen_.compareAndSet(0, 1)) return false;

            subscribeInl(channel);
        }
        return true;
    }

    /**
     * 订阅频道 递归调用
     * @param channel 频道名称数组
     */
    private void subscribeInl(String... channel) {
        thread_.addTask(() -> {
            try{
                SingleUtils.instance(IRedisConnect.class).subscribe(pubSub_, channel);
            } catch (Exception ex) {
                log.warn("channel: ", ex);
                try { Thread.sleep(2000); } catch (Exception ignored){}
                subscribeInl(channel);
            }
        });
    }

    /**
     * 订阅模式
     * @param patten 模式名称数组
     */
    public boolean psubscribe(String... patten) {
        synchronized (listen_) {
            if (!listen_.compareAndSet(0, 2)) return false;
            pSubscribeInl(patten);
        }
        return true;
    }

    /**
     * 订阅模式：递归调用
     * @param patten 模式名称数组
     */
    private void pSubscribeInl(String... patten) {
        thread_.addTask(() -> {
            try {
                SingleUtils.instance(IRedisConnect.class).psubscribe(pubSub_, patten);
            } catch (Exception ex) {
                log.warn("pSubscribe: ", ex);
                try { Thread.sleep(2000); } catch (Exception ignored){}
                pSubscribeInl(patten);
            }
        });
    }

    /**
     * 取消订阅
     */
    public void unsubscribe() {
        synchronized (listen_) {
            if (!listen_.compareAndSet(1, 0)) return;
            try {
                pubSub_.unsubscribe();
            } catch (Exception ignored) {
            }
        }
    }

    /**
     * 取消订阅
     */
    public void punsubscribe() {
        synchronized (listen_) {
            if (!listen_.compareAndSet(2, 0)) return;
            try {
                pubSub_.punsubscribe();
            } catch (Exception ignored) {
            }
        }
    }
}
